package belf.migrate.engine;

import belf.migrate.api.queue.RedisClient;
import belf.migrate.api.source.Source;
import belf.migrate.api.source.cdc.CDCSource;
import belf.migrate.core.util.LoadLogger;
import belf.migrate.engine.admin.SparkServer;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.function.Consumer;

/**
 * 迁移服务器启动类：1.启动任务执行引擎，启动HTTP服务器；2.在虚拟机退出时自动关闭资源
 */
@Slf4j
public class BelfEngine {

    public BelfEngine() {
        LoadLogger.initLog();
    }

    public void start() {
        SparkServer sparkServer = new SparkServer();
        sparkServer.start();
        BelfEvn.getInstance().setHttpServer(sparkServer);

        addShutdownHook();

        log.info("BelfEngine Start Successfully!!!");
    }

    private void addShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> closeEngine()));
    }

    private void closeEngine() {

        // 关闭redis客户端
        RedisClient.getInstance().close();

        // 关闭Spark
        BelfEvn.getInstance().getHttpServer().close();

        // 关闭所有CDC任务
        BelfEvn.getInstance().getCdcSources().forEach(new Consumer<CDCSource>() {
            @Override
            public void accept(CDCSource cdcSource) {
                if (cdcSource != null) { //避免关闭状态不一致造成的空指针
                    cdcSource.stop();
                }
            }
        });

        //关闭所有数据同步任务的写入端Sink
        BelfEvn.getInstance().getJdbcSinks().forEach(sink -> sink.stop());
        //清除所有Job
        BelfEvn.getInstance().stopAll();
        // 关闭所有的目标端Catalog数据库连接
//        belfEvn.getJobContextMap().values().forEach(jobContext -> jobContext.getSinkCatalog().close());
    }

    public static void main(String[] args) {
        BelfEngine engine = new BelfEngine();

        // FIXME: 对于GBase等不支持自动加载驱动的数据库，硬编码在启动migrate时初始化驱动，后续再修复这个问题
        try {
            Class<?> driverClass = Class.forName("com.gbasedbt.jdbc.Driver");
            Constructor<?> constructor = driverClass.getDeclaredConstructor();
            constructor.newInstance();
        } catch (InstantiationException | IllegalAccessException | ClassNotFoundException | NoSuchMethodException |
                 InvocationTargetException e) {
            throw new RuntimeException(e);
        }
        engine.start();
    }
}
